package com.esunny.fptree;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
 * see http://www.fuchaoqun.com/2010/02/recommendation-system-on-association-rules/
 * http://www.cnblogs.com/zhangchaoyang/articles/2198946.html
 * 
 * @author jenvin 2012-7-27 下午12:19:11
 */
public class FPTreeServer {

    private static final Log    log                 = LogFactory.getLog(FPTreeServer.class);
    private DataRetrieveService dataRetrieveService = null;
    private DataStoreService    dataStoreService    = null;
    private int                 minSupport          = 1;
    private int                 threadSize          = 3;
    private String              poolName            = "FPTreeServer";
    private final AtomicInteger threadNumber        = new AtomicInteger(1);
    private ExecutorService     pool                = null;

    public void init() {
        pool = Executors.newFixedThreadPool(threadSize, new ThreadFactory() {

            @Override
            public Thread newThread(Runnable r) {
                Thread t = new Thread(r, poolName + "-thread-" + threadNumber.getAndIncrement());
                return t;
            }
        });
    }

    public void destory() {
        pool.shutdownNow();
    }

    public void start() {
        List<List<String>> records = dataRetrieveService.getRecords();
        if (records == null || records.size() == 0) {
            if (log.isInfoEnabled()) {
                log.info("no records");
            }
            return;
        }

        ExecutionTimer timer = new ExecutionTimer();
        timer.start();
        final FreqItems items = new FreqItems(records, minSupport);
        timer.end();
        if (log.isInfoEnabled()) {
            log.info("build freq items take " + timer.duration() + " ms");
            log.info(items.toString());
        }
        timer.reset();
        timer.start();
        final FreqPrefixTree tree = new FreqPrefixTree(records, items);
        timer.end();
        if (log.isInfoEnabled()) {
            log.info("build freq tree take " + timer.duration() + " ms");
            log.info(tree.toString());// FIXME May be a bit big
        }
        Map<String, Future<List<Result>>> mapResults = new HashMap<String, Future<List<Result>>>();
        for (final Item i : items) {
            Future<List<Result>> futureResult = pool.submit(new Callable<List<Result>>() {

                @Override
                public List<Result> call() throws Exception {
                    List<Result> result = FreqPrefixTrees.findFpResult(tree, i, minSupport);
                    dataStoreService.set(i.getName(), result);
                    return result;
                }
            });
            mapResults.put(i.getName(), futureResult);
        }
        // 等待所有任务结束
        for (Entry<String, Future<List<Result>>> e : mapResults.entrySet()) {
            try {
                e.getValue().get();

                // dataStoreService.set(e.getKey(), result);
            } catch (InterruptedException e1) {
                log.error("error in getting [" + e.getKey() + "] result", e1);
            } catch (ExecutionException e1) {
                log.error("error in getting [" + e.getKey() + "] result", e1);
            }
        }
    }

    public void setDataRetrieveService(DataRetrieveService dataRetrieveService) {
        this.dataRetrieveService = dataRetrieveService;
    }

    public void setDataStoreService(DataStoreService dataStoreService) {
        this.dataStoreService = dataStoreService;
    }

    public void setMinSupport(int minSupport) {
        this.minSupport = minSupport;
    }

    public void setThreadSize(int threadSize) {
        this.threadSize = threadSize;
    }

    public static void main(String[] args) {
        FPTreeServer server = new FPTreeServer();
        server.setDataRetrieveService(new DataRetrieveServiceFileImpl());
        server.setDataStoreService(new DataStoreServicePrintImpl());
        server.setMinSupport(3);
        server.setThreadSize(2);
        server.init();
        server.start();
        server.destory();
    }

}
